package com.yootk.consumer.config;


import com.yootk.consumer.listener.RabbitMQMessageListener;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.PooledChannelConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;

@Configuration
@PropertySource("classpath:config/amqp.properties")
@EnableRabbit
public class RabbitMQConfig {
    @Value("${amqp.rabbitmq.host}")
    private String host;
    @Value("${amqp.rabbitmq.port}")
    private Integer port;
    @Value("${amqp.rabbitmq.username}")
    private String username;
    @Value("${amqp.rabbitmq.password}")
    private String password;
    @Value("${amqp.rabbitmq.vhost}")
    private String vhost;
    @Value("${amqp.rabbitmq.routing.key}")
    private String routingKey;
    @Value("${amqp.rabbitmq.exchange.name}")
    private String exchangeName;
    @Value("${amqp.rabbitmq.queue.name}")
    private String queueName;
    @Bean
    public com.rabbitmq.client.ConnectionFactory amqpConnectionFactory() { // 连接工厂类
        com.rabbitmq.client.ConnectionFactory factory =
                new com.rabbitmq.client.ConnectionFactory();
        factory.setHost(this.host);
        factory.setPort(this.port);
        factory.setUsername(this.username);
        factory.setPassword(this.password);
        factory.setVirtualHost(this.vhost);
        return factory;
    }

    @Bean
    public org.springframework.amqp.rabbit.connection.ConnectionFactory
    springConnectionFactory(
            com.rabbitmq.client.ConnectionFactory amqpConnectionFactory) {
        org.springframework.amqp.rabbit.connection.ConnectionFactory factory; // Spring连接工厂
        factory = new PooledChannelConnectionFactory(
                amqpConnectionFactory);
        return factory;
    }
    @Bean
    public Queue queue(){
        return new Queue(this.queueName,true,false,true) ;
    }
    @Bean
    public RabbitListenerContainerFactory containerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory() ;
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(5);
        factory.setMaxConcurrentConsumers(10);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        int batchSize = 20 ;
        int bufferLimit = 4096 ;
        int timeout = 10000 ;
        BatchingStrategy strategy = new SimpleBatchingStrategy(batchSize,bufferLimit,timeout) ;
        factory.setConsumerBatchEnabled(true);
        factory.setBatchingStrategy(strategy);
        factory.setBatchListener(true);
        return factory ;
    }
    @Bean
    public RetryTemplate template(){
        RetryTemplate template = new RetryTemplate() ;
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy() ;
        policy.setInitialInterval(500);
        policy.setMaxInterval(10000);
        policy.setMultiplier(10.0);
        template.setBackOffPolicy(policy);
        return template ;
    }
    @Bean
    public Exchange exchange(){
        return new FanoutExchange(this.exchangeName) ;
    }
    @Bean
    public Binding binding(Exchange exchangee,Queue queue){
        return BindingBuilder.bind(queue).to(exchangee).with(this.routingKey).noargs() ;
    }
    @Bean
    public RabbitAdmin admin(
            RetryTemplate template,
            Binding binding,
            Exchange exchange,
            Queue queue,
            ConnectionFactory factory
    ){
        RabbitAdmin admin = new RabbitAdmin(factory) ;
        admin.setRetryTemplate(template);
        admin.declareQueue(queue) ;
        admin.declareExchange(exchange);
        admin.declareBinding(binding);
        return admin ;
    }
}
